package com.ecommerce.common.es;

import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.ElasticsearchStatusException;
import org.elasticsearch.client.*;
import com.fasterxml.jackson.core.type.TypeReference;
import org.elasticsearch.rest.RestStatus;

import java.io.IOException;
import java.util.Map;
import java.util.Optional;

/**
 * PIT操作工具类（兼容ES 7.16.2）
 */
@Slf4j
public class PitManager {

    private static final ObjectMapper mapper = new ObjectMapper();

    /**
     * 续期 PIT（兼容 Elasticsearch 7.16.2）
     * @param client 低级客户端
     * @param pitId 需要续期的 PIT ID
     * @param keepAlive 续期时间（如 "5m"）
     * @return 续期后的 PIT ID（可能与原ID相同或不同），若PIT已过期，则返会null
     * @throws IOException 当请求失败时抛出
     */
    public static String extendPit(RestClient client, String pitId, String keepAlive)
            throws IOException {

        // 1. 构建请求
        Request request = new Request("POST", "/_pit/" + pitId + "/_extend");
        // String json = String.format("{\"id\":\"%s\",\"keep_alive\":\"%s\"}", pitId, keepAlive);
        String json=String.format("{\"keep_alive\":\"%s\"}", keepAlive);
        request.setJsonEntity(json);

        // 2. 执行请求
        Response response;
        try {
            response = client.performRequest(request);
        } catch (ElasticsearchStatusException e) {
            if (e.status() == RestStatus.NOT_FOUND) {
                log.error("pit have expired,pitId:{}",pitId);
                // PIT已过期
                return null;
            }
            
            log.error("extendPit,ElasticsearchStatusException",e);
            throw e;
        }

        // 检查响应状态
        if (response.getStatusLine().getStatusCode() != 200) {
            log.error("Error in extending pit: {}", response.getStatusLine());
        }

        // 3. 解析响应
        Map<String, Object> body = mapper.readValue(
                response.getEntity().getContent(),
                new TypeReference<Map<String, Object>>() {}
        );

        // 4. 提取新ID
        String newPitId = extractPitId(body);

        // 5. 处理ID变更
        handleIdChange(client, pitId, newPitId);

        return newPitId;
    }

    private static String extractPitId(Map<String, Object> body) throws IOException {
        //Elasticsearch 7.16.x 可能返回字段名为 "id"，更高版本可能返回 "pit_id"
        String id = Optional.ofNullable(body.get("id"))
                .orElseGet(() -> body.get("pit_id"))
                .toString();

        if (id == null || id.isEmpty()) {
            throw new IOException("响应中缺少有效的pitId");
        }
        return id;
    }

    /**
     * 如果ID变化，需要关闭旧PIT
     * @param client 低级客户端
     * @param oldId 旧PIT id
     * @param newId 新PIT id
     */
    private static void handleIdChange(RestClient client, String oldId, String newId) {
        if (!oldId.equals(newId)) {
            log.warn("PIT_ID发生变化: 旧ID={}, 新ID={}", oldId, newId);
            try {
                closePit(client, oldId);
                log.info("已清理旧PIT资源: {}", oldId);
            } catch (Exception e) {
                log.warn("旧PIT清理失败: {}", oldId, e);
            }
        }
    }

    /**
     * 关闭PIT
     */
    public static void closePit(RestClient client, String pitId) throws IOException {
        Request request = new Request("DELETE", "/_pit");
        request.setJsonEntity(String.format("{\"id\":\"%s\"}", pitId));
        client.performRequest(request);
    }
}
